def _get_bars(
    symbols: Union[str, List[str]],
    freq: str,
    ref_dt1: datetime,
    ref_dt2: Optional[datetime] = None,
    count: int = 0,
    cols: Union[str, List[str], None] = None,
    fq_ref_date: Optional[datetime.date] = None,
    skip_nan: bool = False,
) -> DataFrame:
    """
    获取 Bar 数据
    :param symbols: 标的代码（列表）
    :param freq: 数据频率，1d 1m ...
    :param ref_dt1: 参考时间 1，取数据的开始时间 / 基准时间
    :param ref_dt2: 参考时间 2，与 `count` 二选一，取数据的截止时间，当 symbols 是单只标的时可用
    :param count: 与 `ref_dt2` 二选一，数据条数，正数从 ref_dt1 向未来取，负数从 ref_dt1 向历史取
    :param cols: 要获取的数据列

    支持的字段列表
        * open
        * high
        * low
        * close
        * volume
        * amount
        * limit_up
        * limit_down
        * pre_close 前一个交易日收盘价
        * factor 复权因子
    """
    if (not ref_dt2 and not count) or (ref_dt2 and count):
        raise ParameterError("must specify one of ref_dt2 and count")
    if isinstance(symbols, str):
        symbols = [symbols]
    # if fq_ref_date and symbols and len(symbols) != 1:
    #     raise ParameterError("fq_ref_date is used for only for symbol")

    if cols:
        if isinstance(cols, str):
            cols = [cols]
        if "symbol" not in cols:
            cols.append("symbol")
        if "dt" not in cols:
            cols.append("dt")
    else:
        cols = None
    if freq.endswith("m"):
        step = timedelta(months=1) if ref_dt2 or count > 0 else timedelta(months=-1)
        path_func = lambda dt: (
            Path(get_option("data_path", "/opt/data"))
            / "quote"
            / "stock_index_minute"
            / dt.strftime("%Y-%m")
        ).as_posix()
        if ref_dt2:
            dt_filter = f"(dt >= {ref_dt1.strftime('%Y%m%d%H%M%S')}) & (dt <= {ref_dt2.strftime('%Y%m%d%H%M%S')})"
        elif count > 0:
            dt_filter = f"(dt >= {ref_dt1.strftime('%Y%m%d%H%M%S')})"
        else:
            dt_filter = f"(dt <= {ref_dt1.strftime('%Y%m%d%H%M%S')})"
    else:
        step = timedelta(years=1) if ref_dt2 or count > 0 else timedelta(years=-1)
        path_func = lambda dt: (
            Path(get_option("data_path", "/opt/data"))
            / "quote"
            / "stock_index_daily"
            / str(dt.year)
        ).as_posix()
        if ref_dt2:
            dt_filter = f"(dt >= {ref_dt1.strftime('%Y%m%d')}) & (dt <= {ref_dt2.strftime('%Y%m%d')})"
        elif count > 0:
            dt_filter = f"(dt >= {ref_dt1.strftime('%Y%m%d')})"
        else:
            dt_filter = f"(dt <= {ref_dt1.strftime('%Y%m%d')})"

    symbol_filetrs = [f"(symbol=='{symbol}')" for symbol in symbols if symbol]

    if symbol_filetrs:
        query_str = dt_filter + " & " + "(" + " | ".join(symbol_filetrs) + ")"
    else:
        query_str = dt_filter

    # if skip_nan:
    #     query_str += " & (~np.isnan(open))"

    if count:
        remain_cnt = int(freq[:-1]) * abs(count)
    else:
        remain_cnt = 1000 * 1000 * 1000

    dfl = []
    current_dt = ref_dt1
    while True:
        if (
            ref_dt2
            and current_dt.year >= ref_dt2.year
            and current_dt.month > ref_dt2.month
        ):
            break
        if current_dt > datetime.now():
            break
        if remain_cnt <= 0:
            break

        try:
            ctb = bcolz.ctable(rootdir=path_func(current_dt), mode="r")
        except (KeyError, ValueError):
            current_dt += step
            continue

        # nda = ctb[query_str]
        try:
            nda = ctb.fetchwhere(
                query_str, out_flavor="numpy", vm="python", outcols=cols
            )
        except (StopIteration, StopAsyncIteration):
            current_dt += step
            continue
        if len(nda) > 0:
            dfl.append(nda)
            remain_cnt -= len(nda)
        current_dt += step

    if not dfl:
        return DataFrame()
    data_frame = (
        DataFrame(np.concatenate(dfl, axis=0))
        .astype({"dt": str})
        .astype({"dt": "datetime64[ns]"})
        .sort_values(by="dt")
    )
    if ref_dt2:
        data_frame = data_frame[
            (data_frame["dt"] >= ref_dt1) & (data_frame["dt"] <= ref_dt2)
        ]
    elif count > 0:
        data_frame = data_frame[(data_frame["dt"] >= ref_dt1)].iloc[:count]
    else:
        data_frame = data_frame[(data_frame["dt"] <= ref_dt1)].iloc[count:]
    data_frame = data_frame.set_index(["symbol", "dt"])
    if not fq_ref_date:
        return data_frame

    for symbol in data_frame.index.get_level_values(0).unique():
        symbol_df = data_frame.loc[symbol]
        try:
            post_factor = symbol_df.loc[symbol_df.index >= fq_ref_date].factor[0]
        except KeyError:
            post_factor = symbol_df.loc[symbol_df.index <= fq_ref_date].factor[-1]
        process_factor = 1.0 / post_factor
        # 复权需要复权的字段
        for col in [
            "open",
            "close",
            "high",
            "low",
            "avg",
            "high_limit",
            "low_limit",
            "pre_close",
        ]:
            if col not in data_frame:
                continue
            data_frame.loc[symbol][col] = data_frame.loc[symbol][col] * process_factor
        if "volume" in data_frame:
            data_frame.loc[symbol]["volume"] = (
                data_frame.loc[symbol][col] / process_factor
            )

    return data_frame
